package com.giraffe.mqtt.service;

import com.alibaba.fastjson.JSONObject;
import com.giraffe.common.redis.service.RedisService;
import com.giraffe.device.api.DeviceService;
import com.giraffe.device.pojo.dto.DeviceDTO;
import com.giraffe.device.pojo.enums.DeviceStatusEnum;
import com.giraffe.mqtt.bean.MqttChannel;
import com.giraffe.mqtt.bean.SendMqttMessage;
import com.giraffe.mqtt.bean.SessionMessage;
import com.giraffe.mqtt.bean.WillMeaasge;
import com.giraffe.mqtt.broker.pojo.constants.DeviceCashKeyConstant;
import com.giraffe.mqtt.enums.ConfirmStatus;
import com.giraffe.mqtt.enums.SessionStatus;
import com.giraffe.mqtt.enums.SubStatus;
import com.giraffe.mqtt.utils.ByteBufUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.util.Collection;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author ldx
 * @Date 2024/01/15 12:29
 **/
@Service
@Slf4j
public class MqttService{

    @Resource
    ClientSessionService clientSessionService;
    @Resource
    PublishApiService publishApiService;
    @Resource
    ChannelService channelService;
    @Resource
    WillService willService;
    @Resource
    RedisService redisService;

    @Resource
    DeviceService deviceService;
    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;
    protected ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()*2);

    /**
     * 取消订阅
     */
    public void unsubscribe(String deviceId, List<String> topics1) {
        Optional.ofNullable(channelService.getMqttChannel(deviceId)).ifPresent(mqttChannel -> {
            topics1.forEach(topic -> {
                channelService.deleteChannel(topic,mqttChannel);
            });
        });
    }

    /**
     * 登录成功后 回复
     */
    /**
     * 登录成功后 回复
     */
    public void loginSuccess(String deviceId,Channel channel, MqttConnectMessage mqttConnectMessage) {
        MqttFixedHeader mqttFixedHeader1 = mqttConnectMessage.fixedHeader();
        MqttConnectVariableHeader mqttConnectVariableHeader = mqttConnectMessage.variableHeader();
        final MqttConnectPayload payload = mqttConnectMessage.payload();

        MqttChannel build = buildMqttChannel(deviceId,channel,mqttConnectVariableHeader,payload);
        // 连接成功后调用connectSuccess方法
        connectSuccess(payload.clientIdentifier(), build);

        handleWillMessage(deviceId,payload,mqttConnectVariableHeader,build);

        MqttConnectReturnCode connectReturnCode = MqttConnectReturnCode.CONNECTION_ACCEPTED;
        MqttConnAckVariableHeader mqttConnAckVariableHeader = new MqttConnAckVariableHeader(connectReturnCode, !mqttConnectVariableHeader.isCleanSession());
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(
                MqttMessageType.CONNACK, mqttFixedHeader1.isDup(), MqttQoS.AT_MOST_ONCE, mqttFixedHeader1.isRetain(), 0x02);
        MqttConnAckMessage connAck = new MqttConnAckMessage(mqttFixedHeader, mqttConnAckVariableHeader);
        channel.writeAndFlush(connAck);

        sendSessionMessages(deviceId,payload,build);
    }

    /**
     * 构建MqttChannel对象
     * @param deviceId 设备ID
     * @param channel Channel对象
     * @param mqttConnectVariableHeader MQTT连接变量头部
     * @param payload MQTT连接负载
     * @return 构建的MqttChannel对象
     */
    private MqttChannel buildMqttChannel(String deviceId, Channel channel, MqttConnectVariableHeader mqttConnectVariableHeader, MqttConnectPayload payload) {
        // 使用MqttChannel.builder()创建MqttChannel对象
        MqttChannel build = MqttChannel.builder().channel(channel).cleanSession(mqttConnectVariableHeader.isCleanSession())
                .deviceId(payload.clientIdentifier())
                .sessionStatus(SessionStatus.OPEN)
                .isWill(mqttConnectVariableHeader.isWillFlag())
                .subStatus(SubStatus.NO)
                .topic(new CopyOnWriteArraySet<>())
                .message(new ConcurrentHashMap<>())
                .receive(new CopyOnWriteArraySet<>())
                .index(new AtomicInteger(1))
                .build();
        // 返回构建的MqttChannel对象
        return build;
    }

    /**
     * 处理遗嘱消息
     * @param deviceId 设备ID
     * @param payload MQTT连接负载
     * @param mqttConnectVariableHeader MQTT连接变量头部
     * @param mqttChannel MQTT通道
     */
    private void handleWillMessage(String deviceId, MqttConnectPayload payload, MqttConnectVariableHeader mqttConnectVariableHeader, MqttChannel mqttChannel) {
        if (mqttConnectVariableHeader.isWillFlag()) { // 遗嘱消息标志
            boolean b = payload.willTopic() != null && payload.willMessage() != null;
            if (!b) {
                throw new RuntimeException("will message and will topic is not null");
            }
            // 处理遗嘱消息
            final WillMeaasge buildWill = WillMeaasge.builder().
                    qos(mqttConnectVariableHeader.willQos())
                    .willMessage(deviceId)
                    .willTopic(payload.willTopic())
                    .isRetain(mqttConnectVariableHeader.isWillRetain())
                    .build();
            willService.save(payload.clientIdentifier(), buildWill);
        } else {
            willService.del(payload.clientIdentifier());
            boolean b = !mqttConnectVariableHeader.isWillRetain() && mqttConnectVariableHeader.willQos() == 0;
            if (!b) {
                throw new RuntimeException("will retain should be  null and will QOS equal 0");
            }
        }
    }

    /**
     * 发送 session 数据
     * @param deviceId 设备ID
     * @param payload MQTT连接负载
     * @param mqttChannel MQTT通道
     */
    private void sendSessionMessages(String deviceId, MqttConnectPayload payload, MqttChannel mqttChannel) {
        // 获取 session 数据
        ConcurrentLinkedQueue<SessionMessage> sessionMessages = clientSessionService.getByteBuf(payload.clientIdentifier());
        if(!ObjectUtils.isEmpty(sessionMessages)){
            // 根据 QoS 发送消息
            switch (sessionMessages.poll().getQoS()) {
                case EXACTLY_ONCE:
                    // 发送 QoS 1 确认消息
                    publishApiService.sendQosConfirmMsg(MqttQoS.EXACTLY_ONCE,channelService.getMqttChannel(deviceId), sessionMessages.poll().getTopic(), sessionMessages.poll().getByteBuf());
                    break;
                case AT_MOST_ONCE:
                    // 发送 QoS 0 消息
                    publishApiService.sendQos0Msg(mqttChannel.getChannel(), sessionMessages.poll().getTopic(), sessionMessages.poll().getByteBuf());
                    break;
                case AT_LEAST_ONCE:
                    // 发送 QoS 2 确认消息
                    publishApiService.sendQosConfirmMsg(MqttQoS.AT_LEAST_ONCE,channelService.getMqttChannel(deviceId), sessionMessages.poll().getTopic(), sessionMessages.poll().getByteBuf());
                    break;
            }
        }
    }

    /**
     * qos2 第二步
     */
    public void doPubrel(Channel channel, int messageId) {
        String deviceId = channelService.getDeviceId(channel);
        MqttChannel mqttChannel = channelService.getMqttChannel(deviceId);
        if(mqttChannel.isLogin()){
            mqttChannel.removeRecevice(messageId);
            publishApiService.sendToPubComp(channel,messageId);
        }
    }



    /**
     * qos2 第三步
     */
    
    public void doPubrec(Channel channel, int mqttMessage) {
        publishApiService.sendPubRel(channel,false,mqttMessage);
    }

    /**
     * 连接成功后
     * @param deviceId
     * @param build
     */

    public boolean connectSuccess(String deviceId, MqttChannel build) {
        setDeviceStatus(deviceId, DeviceStatusEnum.ONLINE.getCode());
        return  Optional.ofNullable(channelService.getMqttChannel(deviceId))
                .map(mqttChannel -> {
                    switch (mqttChannel.getSessionStatus()){
                        case OPEN:
                            return false;
                        case CLOSE:
                            switch (mqttChannel.getSubStatus()){
                                case YES: // 清除订阅  topic
                                    deleteSubTopic(mqttChannel).stream()
                                            .forEach(s -> channelService.putCashMapData(s,build));
                                    break;
                            }
                    }
                    channelService.saveMqttChannel(deviceId,build);
                    return true;
                }).orElseGet(() -> {
                    channelService.saveMqttChannel(deviceId,build);
                    return  true;
                });
    }


    /**
     * 订阅成功后 (发送保留消息)
     */
    public void suscribeSuccess(String deviceId, Set<String> topics){
        if(!CollectionUtils.isEmpty(topics)){
            MqttChannel mqttChannel = channelService.getMqttChannel(deviceId);
            mqttChannel.setSubStatus(SubStatus.YES); // 设置订阅主题标识
            mqttChannel.addTopic(topics);
            redisService.setCacheSet(DeviceCashKeyConstant.DEVICE_TOPIC_KEY+deviceId,topics);
            executorService.execute(() -> {
                Optional.ofNullable(mqttChannel).ifPresent(mqttChannel1 -> {
                    if(mqttChannel1.isLogin()){
                        topics.parallelStream().forEach(topic -> {
                            channelService.addChannel(topic,mqttChannel);
                            //sendRetain(topic,mqttChannel); // 发送保留消息
                        });
                    }
                });
            });
        }
    }

    /**
     * 发布消息成功 ()
     * @param channel
     * @param mqttPublishMessage
     */
    
    public void publishSuccess(Channel channel, MqttPublishMessage mqttPublishMessage) {
        MqttFixedHeader mqttFixedHeader = mqttPublishMessage.fixedHeader();
        MqttPublishVariableHeader mqttPublishVariableHeader = mqttPublishMessage.variableHeader();
        MqttChannel mqttChannel = channelService.getMqttChannelByChannel(channel);
        ByteBuf payload = mqttPublishMessage.payload();
        byte[] bytes = ByteBufUtil.copyByteBuf(payload); //
        String message = new String(bytes);
        int messageId = mqttPublishVariableHeader.messageId();
        JSONObject obj = new JSONObject();
        obj.put("deviceId",mqttChannel.getDeviceId());
        obj.put("topic",mqttPublishVariableHeader.topicName());
        obj.put("message",message);
        kafkaTemplate.send("mqtt_data_topic_up",obj.toString());
        executorService.execute(() -> {
            if (channel.hasAttr(channelService.getLOGIN()) && mqttChannel != null) {
                boolean isRetain;
                switch (mqttFixedHeader.qosLevel()) {
                    case AT_MOST_ONCE: // 至多一次
                        break;
                    case AT_LEAST_ONCE:
                        publishApiService.sendPubBack(channel, messageId);
                        break;
                    case EXACTLY_ONCE:
                        publishApiService.sendPubRec(mqttChannel, messageId);
                        break;
                }
                if ((isRetain=mqttFixedHeader.isRetain()) && mqttFixedHeader.qosLevel() != MqttQoS.AT_MOST_ONCE) { //是保留消息  qos >0
                    /*saveRetain(mqttPublishVariableHeader.topicName(),
                            RetainMessage.builder()
                                    .byteBuf(bytes)
                                    .qoS(mqttFixedHeader.qosLevel())
                                    .build(), false);*/
                } else if (mqttFixedHeader.isRetain() && mqttFixedHeader.qosLevel() == MqttQoS.AT_MOST_ONCE) { // 是保留消息 qos=0  清除之前保留消息 保留现在
                    /*saveRetain(mqttPublishVariableHeader.topicName(),
                            RetainMessage.builder()
                                    .byteBuf(bytes)
                                    .qoS(mqttFixedHeader.qosLevel())
                                    .build(), true);*/
                }
                if (!mqttChannel.checkRecevice(messageId)) {
                    push(mqttPublishVariableHeader.topicName(), mqttFixedHeader.qosLevel(), bytes,isRetain);
                    mqttChannel.addRecevice(messageId);
                }
            }
        });

    }
    /**
     * 推送消息给订阅者
     */
    public void push(String topic, MqttQoS qos, byte[] bytes, boolean isRetain){
        Collection<MqttChannel> subChannels = channelService.getChannels(topic, topic1 -> channelService.getCashMapData(topic1));
        if(!CollectionUtils.isEmpty(subChannels)){
            subChannels.parallelStream().forEach(subChannel -> {
                switch (subChannel.getSessionStatus()){
                    case OPEN: // 在线
                        if(subChannel.isActive()){ // 防止channel失效  但是离线状态没更改
                            switch (qos){
                                case AT_LEAST_ONCE:
                                    publishApiService.sendQosConfirmMsg(MqttQoS.AT_LEAST_ONCE,subChannel,topic,bytes);
                                    break;
                                case AT_MOST_ONCE:
                                    publishApiService.sendQos0Msg(subChannel.getChannel(),topic,bytes);
                                    break;
                                case EXACTLY_ONCE:
                                    publishApiService.sendQosConfirmMsg(MqttQoS.EXACTLY_ONCE,subChannel,topic,bytes);
                                    break;
                            }
                        }else{
                            if(!subChannel.isCleanSession() & !isRetain){
                                clientSessionService.saveSessionMsg(subChannel.getDeviceId(),
                                        SessionMessage.builder().byteBuf(bytes).qoS(qos).topic(topic).build() );
                                break;
                            }
                        }
                        break;
                    case CLOSE: // 连接 设置了 clean session =false
                        clientSessionService.saveSessionMsg(subChannel.getDeviceId(),
                                SessionMessage.builder().byteBuf(bytes).qoS(qos).topic(topic).build() );
                        break;
                }
            });
        }
    }

    /**
     * 关闭channel 操作
     * @param deviceId
     */
    
    public void closeSuccess(String deviceId,boolean isDisconnect) {
        if(StringUtils.isNotBlank(deviceId)){
            MqttChannel mqttChannel = channelService.getMqttChannel(deviceId);
            Optional.ofNullable(mqttChannel).ifPresent(mqttChannel1 -> {
                mqttChannel1.setSessionStatus(SessionStatus.CLOSE); // 设置关闭
                mqttChannel1.close(); // 关闭channel
                mqttChannel1.setChannel(null);
                if(!mqttChannel1.isCleanSession()){ // 保持会话
                    // 处理 qos1 未确认数据
                    ConcurrentHashMap<Integer, SendMqttMessage> message = mqttChannel1.getMessage();
                    Optional.ofNullable(message).ifPresent(integerConfirmMessageConcurrentHashMap -> {
                        integerConfirmMessageConcurrentHashMap.forEach((integer, confirmMessage) -> {
                                    if(confirmMessage.getConfirmStatus()== ConfirmStatus.PUB){
                                        clientSessionService.saveSessionMsg(mqttChannel.getDeviceId(), SessionMessage.builder()
                                                .byteBuf(confirmMessage.getByteBuf())
                                                .qoS(confirmMessage.getQos())
                                                .topic(confirmMessage.getTopic())
                                                .build()); // 把待确认数据转入session中
                                    }
                                });

                    });
                }else{  // 删除sub topic-消息
                    channelService.removeMqttChannel(deviceId); // 移除channelId  不保持会话 直接删除  保持会话 旧的在重新connect时替换
                    switch (mqttChannel1.getSubStatus()){
                        case YES:
                            deleteSubTopic(mqttChannel1);
                            break;
                    }
                }
                if(mqttChannel1.isWill()){     // 发送遗言
                    if(!isDisconnect){ // 不是disconnection操作
                        willService.doSend(deviceId);
                    }
                }
            });
            setDeviceStatus(deviceId, DeviceStatusEnum.OFFLINE.getCode());
        }
    }

    /**
     * 清除channel 订阅主题
     * @param mqttChannel
     */
    public Set<String>  deleteSubTopic(MqttChannel mqttChannel){
        Set<String> topics = mqttChannel.getTopic();
        topics.parallelStream().forEach(topic -> channelService.deleteChannel(topic,mqttChannel));
        return topics;
    }

    /**
     * 发送 遗嘱消息(有的channel 已经关闭 但是保持了 session  此时加入session 数据中 )
     * @param willMeaasge 遗嘱消息
     */
    public void sendWillMsg(WillMeaasge willMeaasge){
        Collection<MqttChannel> mqttChannels = channelService.getChannels(willMeaasge.getWillTopic(), topic -> channelService.getCashMapData(topic));
        if(!CollectionUtils.isEmpty(mqttChannels)){
            mqttChannels.forEach(mqttChannel -> {
                switch (mqttChannel.getSessionStatus()){
                    case CLOSE:
                        clientSessionService.saveSessionMsg(mqttChannel.getDeviceId(),
                                SessionMessage.builder()
                                        .topic(willMeaasge.getWillTopic())
                                        .qoS(MqttQoS.valueOf(willMeaasge.getQos()))
                                        .byteBuf(willMeaasge.getWillMessage().getBytes()).build());
                        break;
                    case OPEN:
                        publishApiService.writeWillMsg(mqttChannel,willMeaasge);
                        break;
                }
            });
        }
    }

    private Boolean setDeviceStatus(String deviceId, String deviceStatus){
        DeviceDTO dto = new DeviceDTO();
        dto.setDeviceId(deviceId);
        dto.setDeviceStatus(deviceStatus);
        return deviceService.setDeviceStatus(dto);
    }

    /**
     * 保存保留消息
     * @param topic 主题
     * @param retainMessage 信息
     */
    /*private void saveRetain(String topic, RetainMessage retainMessage, boolean isClean){
        ConcurrentLinkedQueue<RetainMessage> retainMessages = retain.getOrDefault(topic, new ConcurrentLinkedQueue<>());
        if(!retainMessages.isEmpty() && isClean){
            retainMessages.clear();
        }
        boolean flag;
        do{
            flag = retainMessages.add(retainMessage);
        }
        while (!flag);
        retain.put(topic, retainMessages);
    }*/

    /**
     * 发送保留消息
     */
    /*public  void sendRetain(String topic,MqttChannel mqttChannel){
        retain.forEach((_topic, retainMessages) -> {
            if(StringUtils.startsWith(_topic,topic)){
                Optional.ofNullable(retainMessages).ifPresent(pubMessages1 -> {
                    retainMessages.parallelStream().forEach(retainMessage -> {
                        log.info("【发送保留消息】"+mqttChannel.getChannel().remoteAddress()+":"+retainMessage.getString()+"【成功】");
                        switch (retainMessage.getQoS()){
                            case AT_MOST_ONCE:
                                publishApiService.sendQos0Msg(mqttChannel.getChannel(),_topic,retainMessage.getByteBuf());
                                break;
                            case AT_LEAST_ONCE:
                                sendQosConfirmMsg(MqttQoS.AT_LEAST_ONCE,mqttChannel,_topic,retainMessage.getByteBuf());
                                break;
                            case EXACTLY_ONCE:
                                sendQosConfirmMsg(MqttQoS.EXACTLY_ONCE,mqttChannel,_topic,retainMessage.getByteBuf());
                                break;
                        }
                    });
                });
            }
        });

    }*/

}
